package cn.com.lyb.flink.udf;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/*
函数类”也是DataStream API提供的一个函数类的接口，所有的Flink函数类都有其Rich版本。富函数类一般是以抽象类的形式出现的。例如：RichMapFunction、RichFilterFunction、RichReduceFunction等。
与常规函数类的不同主要在于，富函数类可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。
Rich Function有生命周期的概念。典型的生命周期方法有：
open()方法，是Rich Function的初始化方法，也就是会开启一个算子的生命周期。当一个算子的实际工作方法例如map()或者filter()方法被调用之前，open()会首先被调用。
close()方法，是生命周期中的最后一个调用的方法，类似于结束方法。一般用来做一些清理工作。
需要注意的是，这里的生命周期方法，对于一个并行子任务来说只会调用一次；而对应的，实际工作方法，例如RichMapFunction中的map()，在每条数据到来后都会触发一次调用。
 */
public class RichFunctionDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.setParallelism(1);

        DataStreamSource<String> stringDataStreamSource = env.socketTextStream("172.32.1.6", 1234);

        stringDataStreamSource.map(new RichMapFunction<String, Integer>() {
            public void open(Configuration parameters) throws Exception {
                System.out.println("索引是：" + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期开始");
                super.open(parameters);
            }

            @Override
            public void close() throws Exception {
                System.out.println("索引是：" + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期结束");
                super.close();
            }

            @Override
            public Integer map(String value) throws Exception {
                return Integer.parseInt(value) + 1;
            }
        }).print();

//        DataStreamSource<Integer> source = env.fromElements(1, 2, 3, 4);
//        // RichxxxFunction 富函数
//        // 1 多了生命周期管理方法
//        //   open() 每个子任务，在启动时调用一次
//        //   close() 每个子任务，在结束时调用一次
//        // 2 多了一个 运行时上下文
//        //  可以获取一些运行时的环境信息， 比如 子任务编号, 名称
//        source.map(new RichMapFunction<Integer, Integer>() {
//            @Override
//            public void open(Configuration parameters) throws Exception {
//                System.out.println("索引是：" + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期开始");
//                super.open(parameters);
//            }
//
//            @Override
//            public void close() throws Exception {
//                System.out.println("索引是：" + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期结束");
//                super.close();
//            }
//
//            @Override
//            public Integer map(Integer value) throws Exception {
//                return value + 1;
//            }
//        }).print();
        env.execute();
    }
}
